package icu.wdm.flinkcdc.mysql;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.*;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
public class MysqlSink extends RichSinkFunction<String> {

    private static ConcurrentHashMap<String, RefCountedReentrantLock> tableLocks = new ConcurrentHashMap<>();
    private static ExecutorService executorService = Executors.newFixedThreadPool(10); // 可以根据需要调整线程池大小
    private static ExecutorService taskService = Executors.newFixedThreadPool(5); // 可以根据需要调整线程池大小
    private static ConcurrentHashMap<String, BlockingQueue<Runnable>> waitingQueues = new ConcurrentHashMap<>();

    // 初始化方法
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        MySqlDBUtils.initConnectionPool();
    }

    @Override
    public void invoke(String value, Context context) throws Exception {
        JSONObject obj = JSONObject.parseObject(value);
        String op = "";
        if (obj.containsKey("op")) {
            op = obj.getString("op");
        } else if (obj.containsKey("source")) {
            op = "ddl";
        }

        if (op.equals("c") || op.equals("u") || op.equals("r")) {
            handleDmlEvent(obj);
        } else if (op.equals("d")) {
            handleDeleteEvent(obj);
        } else if (op.equals("ddl")) {
            CompletableFuture.runAsync(()->{
                try {
                    handleDdlEvent(obj.getJSONObject("source").getString("table"),obj.getJSONObject("historyRecord"));
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            },taskService);

        } else {
            log.info(">>>>>>> 当前只处理 增CREATE、改UPDATE、删DELETE 和 DDL 事件");
        }
    }

    private void handleDmlEvent(JSONObject obj) throws Exception {
        JSONObject afterObj = obj.getJSONObject("after");
        JSONObject source = obj.getJSONObject("source");
        String tableName = source.getString("table");

        // 检查表是否被锁定
        RefCountedReentrantLock lock = tableLocks.get(tableName);
        if (lock != null && lock.isLocked()) {
            // 表被锁定，加入等待队列
            addTaskToWaitingQueue(tableName, () -> {
                try {
                    executeDmlEvent(afterObj, source);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        } else {
            // 表未被锁定，直接执行
            executeDmlEvent(afterObj, source);
        }
    }

    private void executeDmlEvent(JSONObject afterObj, JSONObject source) throws Exception {
        String tableName = source.getString("table");

        // 用于存储列名和值的列表
        List<String> columns = new ArrayList<>();
        List<Object> values = new ArrayList<>();
        List<String> updates = new ArrayList<>();

        for (Map.Entry<String, Object> entry : afterObj.entrySet()) {
            String key = entry.getKey();
            Object valObj = entry.getValue();
            columns.add("`" + key + "`");
            values.add(valObj);
            updates.add("`" + key + "`=?");
        }

        // 构建SQL语句
        StringBuilder sqlBuilder = new StringBuilder();
        sqlBuilder.append("INSERT INTO ").append(tableName).append(" (")
                .append(String.join(",", columns)).append(") VALUES (")
                .append(String.join(",", Collections.nCopies(values.size(), "?"))).append(") ")
                .append("ON DUPLICATE KEY UPDATE ")
                .append(String.join(",", updates));

        String sql = sqlBuilder.toString();

        if (StringUtils.isNotEmpty(sql)) {
            log.info(">>>>>>>>>>>>插入|更新sql : " + sql);
            log.info("            插入|更新values : " + values);
            int upSert = 0; // 传递两次values，一次用于INSERT，一次用于UPDATE
            try {
                upSert = MySqlDBUtils.executeSql(sql, values, values);
            } catch (SQLException e) {
                log.info("            插入|更新   ->>>失败<<<");
            }
            log.info("            插入|更新   ->>>" + (upSert > 0 ? "成功" : "失败") + "<<<");
        }
    }

    private void handleDeleteEvent(JSONObject obj) throws Exception {
        JSONObject beforeObj = obj.getJSONObject("before");
        JSONObject source = obj.getJSONObject("source");
        String tableName = source.getString("table");

        // 检查表是否被锁定
        RefCountedReentrantLock lock = tableLocks.get(tableName);
        if (lock != null && lock.isLocked()) {
            // 表被锁定，加入等待队列
            addTaskToWaitingQueue(tableName, () -> {
                try {
                    executeDeleteEvent(beforeObj, source);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        } else {
            // 表未被锁定，直接执行
            executeDeleteEvent(beforeObj, source);
        }
    }

    private void executeDeleteEvent(JSONObject beforeObj, JSONObject source) throws Exception {
        String tableName = source.getString("table");
        String id = beforeObj.getString("id");
        String sql = "DELETE FROM " + tableName + " WHERE id='" + id + "'";

        if (StringUtils.isNotEmpty(sql)) {
            log.info(">>>>>>>>>>>>删除sql : " + sql);
            int deleteCount = 0;
            try {
                deleteCount = MySqlDBUtils.executeSql(sql, null,null);
            } catch (SQLException e) {
                log.info(">>>>>>>>>>>>删除   ->>>>>>>失败<<<");
            }
            log.info(">>>>>>>>>>>>删除   ->>>>>>>" + (deleteCount > 0? "成功" : "失败") + "<<<");
        }
    }

    private void handleDdlEvent(String tableName,JSONObject obj) throws Exception {
        String ddl = obj.getString("ddl");

        // 获取表的锁
        RefCountedReentrantLock lock = tableLocks.computeIfAbsent(tableName, k -> new RefCountedReentrantLock());
        lock.lock();

        try {
            // 移除 DDL 语句中的数据库名
            ddl = removeDatabaseNameFromDdl(ddl);
            log.info("----------->DDL事件sql : " + ddl);

            // 执行 DDL 语句
            Thread.sleep(1000 * 50);
            int result = 0;
            try {
                result = MySqlDBUtils.executeSql(ddl,null,null);
            } catch (SQLException e) {
                log.info(">>>>>>>>>>>>执行DDL语句   ->>>失败<<<");
            }
            log.info("----------->执行DDL语句->>>" + (result >= 0 ? "成功" : "失败") + "<<<");
        } finally {
            lock.unlock();
            if (!lock.isLocked()) {
                // 释放锁后，处理等待队列中的任务
                processWaitingQueue(tableName);
            }

        }
    }

    private String removeDatabaseNameFromDdl(String ddl) {
        // 使用正则表达式移除 DDL 语句中的数据库名
        return ddl.replaceAll("`\\w+`\\.", "");
    }

    private void addTaskToWaitingQueue(String tableName, Runnable task) {
        waitingQueues.computeIfAbsent(tableName, k -> new LinkedBlockingQueue<>()).add(task);
    }

    private void processWaitingQueue(String tableName) {
        BlockingQueue<Runnable> queue = waitingQueues.get(tableName);
        if (queue != null) {
            while (!queue.isEmpty()) {
                Runnable task = queue.poll();
                if (task != null) {
                    CompletableFuture.runAsync(task, executorService);
                }
            }
        }
    }

    @Override
    public void close() throws Exception {
        super.close();
        MySqlDBUtils.closeConnectionPool();
        executorService.shutdown();
    }
}